package com.etl.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.Objects;

@Slf4j
@Configuration
public class ElasticSearchConfig {

    /**
     * 权限验证
     */
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

    /**
     * 使用冒号隔开ip和端口
     */
    @Value("${spring.data.elasticsearch.http-host-port}")
    private String[] httpHostPort;

    /**
     * RestClientBuilder 实例
     *
     * @return RestClientBuilder
     */
    @Bean
    public RestClientBuilder restClientBuilder() {
        HttpHost[] hosts = Arrays.stream(httpHostPort)
                .map(this::buildHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        log.info("RestClientBuilder: {}", Arrays.toString(hosts));
        // 配置权限验证
        // credentialsProvider.setCredentials(
        // AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        return RestClient.builder(hosts)
                .setHttpClientConfigCallback(httpClientBuilder ->
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
    }

    /**
     * RestHighLevelClient实例
     *
     * @param restClientBuilder RestClientBuilder
     * @return RestHighLevelClient
     */
    @Bean(name = "highLevelClient")
    public RestHighLevelClient highLevelClient(@Autowired RestClientBuilder restClientBuilder) {
        return new RestHighLevelClient(restClientBuilder);
    }

    /**
     * 建造HttpHost对象
     *
     * @param httpHostString 字符串
     * @return HttpHost
     */
    private HttpHost buildHttpHost(String httpHostString) {
        assert StringUtils.isNotEmpty(httpHostString);
        String[] hostnameAndPort = httpHostString.split(":");
        if (hostnameAndPort.length == 2) {
            String hostname = hostnameAndPort[0];
            int port = Integer.parseInt(hostnameAndPort[1]);
            return new HttpHost(hostname, port);
        }
        return null;
    }

}
